package com.mxnavi5.example.DataStream;

import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.concurrent.TimeUnit;

public class MyEnvTest {
    public static void main(String[] args) throws Exception {
    //    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        Configuration configuration = new Configuration();
        configuration.setInteger(RestOptions.PORT,8081);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(configuration);

        //禁用算子链合并优化，用于性能测试观察算子反压情况
//        env.disableOperatorChaining();

        //设置处理时间的时间语义   处理时间，事件事件
        env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime);

        //设置并行度,集群模式下默认值为配置parallelism,default的值；在本地模式下：默认的并行度为CPU核数可以执行线程数的最大值
        env.setParallelism(1);

//        //设置重启策略  重启3次  每次等待5秒
//        //1）固定延迟重启策略（Fixed Delay Restart Strategy） aly
//        //2）故障率重启策略（Failure Rate Restart Strategy） aly
//        //3）没有重启策略（No Restart Strategy） aly
//        //4）后背重启策略（Fallback Restart Strategy）
//        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, Time.of(5, TimeUnit.SECONDS)));
//
//        //savePoint  设置默认的savePoint路径   一般都是执行savePoint时指定路径
//        env.setDefaultSavepointDirectory("file:///tmp/savepoints");
//
//        //ck策略
//        //开启checkpoint  每分钟一次  精确一次性消费    EXACTLY_ONCE  AT_LEAST_ONCE
//        env.enableCheckpointing(60*1000,CheckpointingMode.EXACTLY_ONCE);
//        // 设置checkpoint的超时时间 即一次checkpoint必须在该时间内完成 不然就丢弃
//        env.getCheckpointConfig().setCheckpointTimeout(10*60*1000);
//        // 设置两次checkpoint之间的最小时间间隔
//        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500);
//        // 设置并发checkpoint的数目
//        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
//        //ck保存路径
//        env.setStateBackend(new FsStateBackend("file:///tmp/checkpoints"));


        SingleOutputStreamOperator dataStreamSource = env.addSource(new UdfSourceTest()).returns(Integer.class)
                .map(x->x.toString()).returns(String.class);
        dataStreamSource.print("gogogo");


        env.execute();


    }

}
